/*
 * =====================================================================================
 *
 *       Filename:  cl_thread_pool.hxx
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  04/23/2011 07:03:44 AM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Chen Ken (Ken), zlchen.ken@gmail.com
 *        Company:  EMC
 *
 * =====================================================================================
 */

#ifndef _CL_THREAD_POOL_HXX_ 
#define _CL_THREAD_POOL_HXX_

#include <memory>
#include <functional>
#include <atomic>
#include <vector>
#include "cl_callback.hxx"
#include "cl_pc_queue.hxx"
#include "cl_thread.hxx"
#include "cl_utils.hxx"

namespace neo
{

class CLThreadPool
{
public:
  enum SchedulePolicy
  {
    ROUND_ROBIN,
    RANDOM,
  };

public:
  explicit CLThreadPool(int num = 0, CLThreadPool::SchedulePolicy policy = RANDOM);
  void start(void);
  void run(const CLCallbackFunctor& cb);
  void shutdown(void);
  
private:
struct CLPoolThread
{
public:
  CLPoolThread(int qindex, CLThreadPool* thrPool) 
    : m_jobQindex(qindex),
    m_thrPool(thrPool)
  {
    m_thr = std::unique_ptr<CLThread>(new CLThread(std::bind(&CLPoolThread::run, this)));
  }
  
  void start(void)
  {
    m_thr->start();
  }
  
  void run(void)
  {
    CLCallbackFunctor cb;
    for(;!m_shutdown;)
    {
      if(m_thrPool->m_jobQueues[m_jobQindex]->pop(cb)) 
      {
        cb();
      }
    }
  }
   
  void shutdown(void)
  {
    m_shutdown = true;
  }
  
  void join(void)
  {
    m_thr->join();
  }

private:
  std::unique_ptr<CLThread> m_thr;
  std::atomic<bool> m_shutdown;
  int m_jobQindex;
  CLThreadPool* m_thrPool;
};

private:
  typedef std::unique_ptr<CLPoolThread> PoolThreadUPtr;
  typedef std::vector<PoolThreadUPtr> JobThreadVector; 
  typedef CLProducerConsumerQueue<CLCallbackFunctor> JobQueue; 
  typedef std::unique_ptr<JobQueue> JobQueueUPtr;
  typedef std::vector<JobQueueUPtr> JobQueueVector;

private:
  SchedulePolicy m_policy;
  std::atomic<int> m_nextThr;
  JobQueueVector m_jobQueues;
  JobThreadVector m_thrs;
};

} // neo

#endif //_CL_THREAD_POOL_HXX_

